package org.apache.lucene.store;

import static org.apache.lucene.store.GAELuceneConstants.blockSize;
import static org.apache.lucene.store.GAELuceneConstants.entityKind;
import static org.apache.lucene.store.GAELuceneConstants.fileBlobPropertyPrefixName;
import static org.apache.lucene.store.GAELuceneConstants.fileCountPropertyName;
import static org.apache.lucene.store.GAELuceneConstants.fileLastModifyDatePrefixName;
import static org.apache.lucene.store.GAELuceneConstants.fileLengthPrefixName;
import static org.apache.lucene.store.GAELuceneConstants.fileLocationReferenceCountPrefixName;
import static org.apache.lucene.store.GAELuceneConstants.fileLocationReferenceEntityPrefixName;
import static org.apache.lucene.store.GAELuceneConstants.fileLocationReferenceSectionSizePrefixName;
import static org.apache.lucene.store.GAELuceneConstants.fileNamePropertyPrefixName;
import static org.apache.lucene.store.GAELuceneConstants.maxEntityBlocks;
import static org.apache.lucene.store.GAELuceneConstants.versionPropertyName;
import static org.apache.lucene.store.GAELuceneConstants.directorySpansEntities;
import static org.apache.lucene.store.GAELuceneConstants.directoryEntities;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.logging.Logger;

import javax.transaction.TransactionRolledbackException;

import com.google.appengine.api.datastore.Blob;
import com.google.appengine.api.datastore.DatastoreFailureException;
import com.google.appengine.api.datastore.DatastoreService;
import com.google.appengine.api.datastore.DatastoreServiceFactory;
import com.google.appengine.api.datastore.DatastoreTimeoutException;
import com.google.appengine.api.datastore.Entity;
import com.google.appengine.api.datastore.EntityNotFoundException;
import com.google.appengine.api.datastore.Key;
import com.google.appengine.api.datastore.KeyFactory;
import com.google.appengine.api.datastore.Transaction;
import com.google.appengine.api.memcache.ErrorHandler;
import com.google.appengine.api.memcache.MemcacheService;
import com.google.appengine.api.memcache.MemcacheServiceException;
import com.google.appengine.api.memcache.MemcacheServiceFactory;
import com.google.appengine.api.memcache.StrictErrorHandler;
import com.google.apphosting.api.ApiProxy.ApiProxyException;
import com.google.apphosting.api.ApiProxy.CapabilityDisabledException;
import com.google.apphosting.api.ApiProxy.OverQuotaException;
import com.google.apphosting.api.ApiProxy.RequestTooLargeException;

public class GAELuceneStaticMethods {

	private static final ErrorHandler memcacheErrorHandler = new StrictErrorHandler();
	
	/**
	 * Takes the static entityKind, and the directory getLockID to
	 * build a standardized Key for this directory.
	 * 
	 * @param directory
	 * @return
	 */
	public static final Key buildDirectoryKey(Directory directory) {
		return buildDirectoryKey(null, directory);
	}
	
	public static final Key buildDirectoryKey(Key parentKey, Directory directory) {
		return KeyFactory.createKey(parentKey, entityKind, directory.getLockID());
	}
	
	/**
	 * Builds a child key of the directoryKey, that has the same name
	 * as the directoryKey. 
	 * 
	 * Expected use is for storing the version
	 * and spansMultipleEntities fields in an entity other than
	 * the directory itself for quick and low-bandwidth retrieval.
	 * 
	 * @param directoryKey
	 * @return
	 */
	public static final Key buildDirectoryVersionKey(Key directoryKey) {
		return directoryKey.getChild(entityKind, directoryKey.getName());
	}
	
	/**
	 * This method takes the key that will be assigned to the top
	 * level directory itself. This method will determine how many
	 * entities need to be created and stored in order to most
	 * efficiently store the entire directory contents.
	 * 
	 * @param directoryName - Name of the directory
	 * @param fileNamesWithSizes - HashMap associating filenames with their byte[].length
	 * @Param overallSize - The size of all the files put together
	 * @return
	 */
	public static final List<Entity> buildEntities(Directory directory) throws IOException {
		// first we need to synchronized against the directory object itself
		// we can't have something else messing with it while we are trying
		// to read the contents from it.
		synchronized (directory) {
			// some basic variables we need
			String[] filenames = directory.listAll();
			ArrayList<Entity> outputEntities = new ArrayList<Entity>();
			Set<Key> outputEntityKeys = new HashSet<Key>();
			
			// get the directory entity ready
			Entity directoryEntity = new Entity(buildDirectoryKey(directory));
			Entity directoryVersionEntity = new Entity(buildDirectoryVersionKey(directoryEntity.getKey()));
			outputEntities.add(directoryEntity);
			outputEntities.add(directoryVersionEntity);
			
			// put the data into the entities
			int blocksRemainingThisEntity = maxEntityBlocks;
			int entityNumber = 1; // index for each child entity of the directory entity
			Key currentEntityKey = directoryEntity.getKey().getChild(entityKind, entityNumber);
			Entity currentEntity = new Entity(currentEntityKey);
			int fileNumber = 1; // index for each file within the directory
			
			// put the number of files into the directory entity
			directoryEntity.setUnindexedProperty(fileCountPropertyName, filenames.length);
			
			for (String filename : filenames) {
				// calculate the size of the file and how it will be split up
				long fileLength = directory.fileLength(filename);
				long fileModified = directory.fileModified(filename);
				
				// put the available meta-data into directory entity	
				directoryEntity.setUnindexedProperty(fileNamePropertyPrefixName + fileNumber, filename);
				directoryEntity.setUnindexedProperty(fileLastModifyDatePrefixName + fileNumber, fileModified);
				directoryEntity.setUnindexedProperty(fileLengthPrefixName + fileNumber, fileLength);
				
				// calculate the file size information
				long completeFileBlocks = (long)(fileLength / blockSize);
				int remainingBlockSizeInBytes = (int)(fileLength % blockSize);
				boolean partialBlockAtEnd = remainingBlockSizeInBytes > 0 ? true : false;
				long fileBlocks = completeFileBlocks + (partialBlockAtEnd ? 1 : 0);
				
				// get the file reader
				IndexInput fileReader = directory.openInput(filename);
				
				// initialize the loop variables for this file
				long blocksRemainingThisFile = fileBlocks;
				int fileSection = 0; // index for each section of the file that must be stored in another entity
				
				while (blocksRemainingThisFile > 0) {
					// increment file sections
					fileSection ++;
					
					// see how much is left to write either to this entity or of the file
					int blocksThisEntity = (int)(blocksRemainingThisEntity <= blocksRemainingThisFile ? blocksRemainingThisEntity : blocksRemainingThisFile);
					int bytesToWrite = blocksThisEntity < blocksRemainingThisFile ? (blocksThisEntity * blockSize) : ((partialBlockAtEnd ? (blocksThisEntity - 1) : blocksThisEntity) * blockSize) + remainingBlockSizeInBytes;
					byte[] fileDataThisEntity = new byte[bytesToWrite];
					
					// put the data into the entity
					fileReader.readBytes(fileDataThisEntity, 0, bytesToWrite);
					Blob blob = new Blob(fileDataThisEntity);
					currentEntity.setUnindexedProperty(fileBlobPropertyPrefixName + fileNumber, blob);
					
					// put the index data into the directory entity
					directoryEntity.setUnindexedProperty(fileLocationReferenceEntityPrefixName + fileNumber + "_" + fileSection, currentEntityKey);
					directoryEntity.setUnindexedProperty(fileLocationReferenceSectionSizePrefixName + fileNumber + "_" + fileSection, bytesToWrite);
					
					// decrement blocks remaining this file
					blocksRemainingThisFile -= blocksThisEntity;
					blocksRemainingThisEntity -= blocksThisEntity;
					
					// flip to the next entity if the current one is full
					if (blocksRemainingThisEntity == 0) {
						// put the currentEntity reference into the outputEntities
						outputEntities.add(currentEntity);
						outputEntityKeys.add(currentEntityKey);
						// increment the current entity number
						entityNumber ++;
						// reset the blocksRemainingThisEntity variable
						blocksRemainingThisEntity = maxEntityBlocks;
						// replace the currentEntity reference with a new entity reference
						currentEntityKey = directoryEntity.getKey().getChild(entityKind, entityNumber);
						currentEntity = new Entity(currentEntityKey);
					}
				}
				
				// put the number of file number of sections into the directory index
				directoryEntity.setUnindexedProperty(fileLocationReferenceCountPrefixName + fileNumber, fileSection);
				
				// increment the file number counter
				fileNumber ++;
			}
			
			// put the last currentEntity into the outputEntities or merge into the directoryEntity if only one
			if (entityNumber == 1) {
				// put null values into the reference location entity properties
				for (int i = 1; i < fileNumber; i++) {
					// check how many sections there are, will be either 0 or 1
					int sectionsCount = (Integer)directoryEntity.getProperty(fileLocationReferenceCountPrefixName + i);
					// if there was 1, then we need to update the key reference to null
					if (sectionsCount == 1)
						directoryEntity.setUnindexedProperty(fileLocationReferenceEntityPrefixName + i + "_1", null);
					// else should not be necessary, multi-section file would mean there is a bug because there would
					// be multiple entities as well, which means the if statement this for loop in nested into won't execute
				}
				
				// copy the properties from the currentEntity
				directoryEntity.setPropertiesFrom(currentEntity);
				
				// mark that the directory doesn't span
				directoryVersionEntity.setUnindexedProperty(directorySpansEntities, false);
				directoryEntity.setUnindexedProperty(directorySpansEntities, false);
			} else {
				if (currentEntity.getProperties().size() > 0) {
					// add the currentEntity into the list we are sending back to the caller
					outputEntities.add(currentEntity);
					outputEntityKeys.add(currentEntityKey);
				}
				
				// mark that directory does span multiple entities
				directoryVersionEntity.setUnindexedProperty(directorySpansEntities, true);
				directoryEntity.setUnindexedProperty(directorySpansEntities, true);
				
				// set the entity keys set into the directory
				directoryEntity.setUnindexedProperty(directoryEntities, outputEntityKeys);
			}
		
			// return the entities
			return outputEntities;
		}
	}
	
	/**
	 * Sets the version number on the directoryEntity. Does not persist the entity only
	 * puts the value into the entity object.
	 * 
	 * @param directoryEntity
	 * @param version
	 */
	public static final void setDirectoryVersion(Entity directoryEntity, Long version) {
		directoryEntity.setUnindexedProperty(versionPropertyName, version);
	}
	
	/**
	 * Return the version number stored in the directoryEntity. Returns null if one
	 * is not found.
	 * 
	 * @param directoryEntity
	 * @return
	 */
	public static final Long getDirectoryVersion(Entity directoryEntity) {
		return (Long)directoryEntity.getProperty(versionPropertyName);
	}
	
	/**
	 * Generates string that is used by the LockFactory to ensure each directories lock
	 * is unique to that directory and does not collide with any other directories. Note
	 * the if two directories have the same name they are the same directory by this definition
	 * so be sure you do not accidentally name two different directories with the same name.
	 * 
	 * @param directory
	 * @param lockName
	 * @return
	 */
	protected static final String getPrefixedLockName(String directory, String lockName) {
		return directory + "." + lockName;
	}
	
	/**
	 * Returns the number of files contained in the directory. You can use this in a for loop
	 * to read each file out of the directory using the other methods in the class. The fileNumbers
	 * are a 1 based index.
	 * 
	 * @param directoryEntity
	 * @return
	 */
	public static final int getFileCount(Entity directoryEntity) {
		try {
			return ((Number)directoryEntity.getProperty(fileCountPropertyName)).intValue();
		} catch (NullPointerException nullPointerException) {
			throw new IllegalArgumentException();
		}
	}
	
	/**
	 * Returns a list of filenames from the directory entity. The fileNumber will correspond to the
	 * List.Key + 1. File numbers have a one based index where the List has a zero based index. There
	 * is no function to return a list of fileNumbers, but there is a function to return the number of files.
	 * 
	 * @param directoryEntity
	 * @return
	 */
	public static final List<String> getFilenames(Entity directoryEntity) {
		// get the number of files
		Integer files = getFileCount(directoryEntity);
		if (files == null) throw new IllegalArgumentException("The number of files is missing from the entity, this must not be a valid directory entity");
		
		// loop over the files building the list
		List<String> fileList = new ArrayList<String>(files);
		for (int i = 1; i <= files; i++)
			fileList.add((String)directoryEntity.getProperty(fileNamePropertyPrefixName + i));
		
		return fileList;
	}
	
	public static final String getFilename(Entity directoryEntity, int fileNumber) {
		return (String)directoryEntity.getProperty(fileNamePropertyPrefixName + fileNumber);
	}
	
	/**
	 * Returns the lastModifyDate in millis long format for the given fileNumber. The fileNumber is
	 * a 1 based index.
	 * @param directoryEntity
	 * @param fileNumber
	 */
	public static final long getFileLastModifyDate(Entity directoryEntity, int fileNumber) {
		try {
			return ((Number)directoryEntity.getProperty(fileLastModifyDatePrefixName + fileNumber)).longValue();
		} catch (NullPointerException nullPointerException) {
			throw new IllegalArgumentException();
		}
	}
	
	
	/**
	 * Returns the length in bytes of a file within this directory.
	 * 
	 * @param directoryEntity
	 * @param fileNumber
	 * @return
	 */
	public static final long getFileLength(Entity directoryEntity, int fileNumber) {
		try {
			return ((Number)directoryEntity.getProperty(fileLengthPrefixName + fileNumber)).longValue();
		} catch (NullPointerException nullPointerException) {
			throw new IllegalArgumentException();
		}
	}
	
	/**
	 * Returns the number of sections that a given file is split into. Each section will be stored in a different
	 * entity each possibly with a different sizeInBytes length. See @link getFileLocationReferenceEntityKeys and
	 * @link getFileLocationReferenceSectionSizes.
	 * 
	 * @param directoryEntity
	 * @param fileNumber
	 * @return
	 */
	public static final int getFileLocationReferenceCount(Entity directoryEntity, int fileNumber) {
		try {
			return ((Number)directoryEntity.getProperty(fileLocationReferenceCountPrefixName + fileNumber)).intValue();
	} catch (NullPointerException nullPointerException) {
			throw new IllegalArgumentException();
		}
	}
	
	/**
	 * Returns a list of FileLocationReferenceEntityKeys for a given fileNumber in the directory.
	 * 
	 * The Entity Keys are what you use to store and access the data entities in datastore and memcache
	 * they are referenced by a sectionNumber which correlates with SectionSize. A null value indicates
	 * that this section is stored in the directoryEntity itself. See @link getFileLocationReferenceCount and
	 * @link getFileLocationReferenceSectionSizes.
	 * 
	 * @param directoryEntity
	 * @param fileNumber
	 * @return
	 */
	public static final List<Key> getFileLocationReferenceEntityKeys(Entity directoryEntity, int fileNumber) {
		Integer referencesCount = getFileLocationReferenceCount(directoryEntity, fileNumber); 

		List<Key> keys = new ArrayList<Key>(referencesCount);
		for (int i = 1; i <= referencesCount; i++)
			keys.add((Key)directoryEntity.getProperty(fileLocationReferenceEntityPrefixName + fileNumber + "_" + i));
		
		return keys;
	}
	
	public static final Key getFileLocationReferenceEntityKey(Entity directoryEntity, int fileNumber, int sectionNumber) {
		return (Key)directoryEntity.getProperty(fileLocationReferenceEntityPrefixName + fileNumber + "_" + sectionNumber);
	}
	
	/**
	 * Returns a list of FileLocationReferenceSectionSizes for a given fileNumber in the directory.
	 * 
	 * The Section Size is the sizeInBytes that file occupies in the entity with the sectionNumber.
	 * See @link getFileLocationReferenceCount and @link getFileLocationReferenceEntityKeys.
	 * 
	 * @param directoryEntity
	 * @param fileNumber
	 * @return
	 */
	public static final List<Integer> getFileLocationReferenceSectionSizes(Entity directoryEntity, int fileNumber) {
		Integer referencesCount = getFileLocationReferenceCount(directoryEntity, fileNumber);
		
		List<Integer> sectionSizes = new ArrayList<Integer>(referencesCount);
		for (int i = 1; i <= referencesCount; i++)
			sectionSizes.add((Integer)directoryEntity.getProperty(fileLocationReferenceSectionSizePrefixName + fileNumber + "_" + i));
		
		return sectionSizes;
	}
	
	public static final int getFileLocationReferenceSectionSize(Entity directoryEntity, int fileNumber, int sectionNumber) {
		return ((Number)directoryEntity.getProperty(fileLocationReferenceSectionSizePrefixName + fileNumber + "_" + sectionNumber)).intValue();
	}
	
	/**
	 * Returns the boolean indicating if there are multiple entities storing this directory data.
	 * 
	 * @param directoryEntity
	 * @return
	 */
	public static final Boolean getDirectorySpansEntities(Entity directoryEntity) {
		return (Boolean)directoryEntity.getProperty(directorySpansEntities);
	}
	
	/**
	 * Returns the Set of Keys that contain the other entities in the directory.
	 * 
	 * @param directoryEntity
	 * @return
	 */
	@SuppressWarnings("unchecked")
	public static final Collection<Key> getDirectoryEntities(Entity directoryEntity) {
		return (Collection)directoryEntity.getProperty(directoryEntities);
	}
	
	/**
	 * Returns the entity specified by the Key. If useMemcache = true then it will look in memcache first
	 * then check datastore, if it wasn't found in memcache it will be put there (based on the flag as well).
	 * 
	 * Returns null if entity not found.
	 * If the datastore issues a timeout error, the method will retry up to five times. With sleep times of 100ms, 200ms, 400ms, 800ms, 1600ms
	 * 
	 * @param key
	 * @param useMemcache
	 * @return Entity
	 * @throws IOException
	 */
	public static final Entity readEntityFromPersistentStore(Key key, boolean useMemcache) throws IOException {
		return readEntityFromPersistentStore(key, useMemcache, null);
	}
	
	/**
	 * Returns the entity specified by the Key. If useMemcache = true then it will look in memcache first
	 * then check datastore, if it wasn't found in memcache it will be put there (based on the flag as well).
	 * 
	 * If you pass in a transaction, then the method will ignore useMemcache to make the read Serializable on the passed in transaction.
	 * 
	 * Returns null if entity not found.
	 * If the datastore issues a timeout error, the method will retry up to five times. With sleep times of 100ms, 200ms, 400ms, 800ms, 1600ms
	 * 
	 * @param key
	 * @param useMemcache
	 * @param transaction
	 * @return Entity
	 * @throws IOException
	 */
	public static final Entity readEntityFromPersistentStore(Key key, boolean useMemcache, Transaction transaction) throws IOException {
		DatastoreService datastoreService = DatastoreServiceFactory.getDatastoreService();
		MemcacheService memcacheService = useMemcache ? MemcacheServiceFactory.getMemcacheService() : null;
		if (memcacheService != null)
			memcacheService.setErrorHandler(memcacheErrorHandler);
		Entity entity = null;
		useMemcache = useMemcache && transaction != null;
		
		if (useMemcache)
		try {
			// try to read it from memcache
			entity = (Entity)memcacheService.get(key);
		} catch (MemcacheServiceException memcacheServiceException) {} 
		
		// if we don't have a directory entity, we will try to read it from datastore
		if (entity == null) {
			try {
				
				int timeout = 100;
				int attempts = 1;
				while (true) {
					try {
						// read the directory entity from datastore
						entity = datastoreService.get(transaction, key);
						// exit the loop
						break;
					} catch (DatastoreTimeoutException datastoreTimeoutException) {
						if (attempts == 5) throwIOWrappedException(datastoreTimeoutException, "read entity");
						try {
							Thread.sleep(timeout);
						} catch (InterruptedException interruptedException) {}
						timeout *= 2;
						attempts ++;
					}
				}
				// put the value into memcache if the flag is set
				if (useMemcache) 
					try {
						memcacheService.put(key, entity);
					} catch (MemcacheServiceException memcacheServiceException) {}
					
			} catch (EntityNotFoundException entityNotFound) {
				entity = null;
			} catch (RuntimeException runtimeException) {
				throwIOWrappedException(runtimeException, "read entity");
			}
			
		}
		
		return entity;
	}
	
	/**
	 * Returns the entities specified by the Keys. If useMemcache = true then it will look in memcache first
	 * then check datastore, if it wasn't found in memcache it will be put there (based on the flag as well).
	 * 
	 * If the datastore issues a timeout error, the method will retry up to five times. With sleep times of 100ms, 200ms, 400ms, 800ms, 1600ms
	 * 
	 * 
	 * @param keys
	 * @param useMemcache
	 * @return Map<Key, Entity>
	 * @throws IOException
	 */
	public static final Map<Key, Entity> readEntitiesFromPersistentStore(Collection<Key> keys, boolean useMemcache) throws IOException {
		return readEntitiesFromPersistentStore(keys, useMemcache, null);
	}
	
	/**
	 * Returns the entities specified by the Keys. If useMemcache = true then it will look in memcache first
	 * then check datastore, if it wasn't found in memcache it will be put there (based on the flag as well).
	 * 
	 * If you pass in a transaction, then the method will ignore useMemcache to make the read Serializable on the passed in transaction.
	 * 
	 * If the datastore issues a timeout error, the method will retry up to five times. With sleep times of 100ms, 200ms, 400ms, 800ms, 1600ms
	 * 
	 * 
	 * @param keys
	 * @param useMemcache
	 * @param transaction
	 * @return Map<Key, Entity>
	 * @throws IOException
	 */
	public static final Map<Key, Entity> readEntitiesFromPersistentStore(Collection<Key> keys, boolean useMemcache, Transaction transaction) throws IOException {
		DatastoreService datastoreService = DatastoreServiceFactory.getDatastoreService();
		MemcacheService memcacheService = useMemcache ? MemcacheServiceFactory.getMemcacheService() : null;
		if (memcacheService != null)
			memcacheService.setErrorHandler(memcacheErrorHandler);
		Map<Key, Entity> entities = new HashMap<Key, Entity>(keys.size());
		useMemcache = useMemcache && transaction != null;
		
		try {
			// read the entities from memcache
			if (useMemcache) {
				try {
					// deal with some silly compiler issue with casting between generic collections and non-generic collections				
					Map<Object, Object> memcacheEntities = memcacheService.getAll(new ArrayList<Object>(keys));
					for (Object key : memcacheEntities.keySet())
						entities.put((Key)key, (Entity)memcacheEntities.get(key));
				} catch (RuntimeException runtimeException) {}
			}
				
			// read the entities from datastore if some or all of them are not in the map
			if (entities.size() < keys.size()) {
				Collection<Key> datastoreKeys = null;
				if (entities.size() > 0) {
					// put the keys we still need into the list
					datastoreKeys = new ArrayList<Key>(keys.size());
					for (Key key : keys)
						if (!entities.containsKey(key))
							datastoreKeys.add(key);
				} else
					// make the datastoreKeys reference just a reference to the passed list object
					datastoreKeys = keys;
				
				// perform the actual read
				Map<Key, Entity> datastoreResults = null;
				int attempts = 1;
				int timeout = 100;
				while (true) 
				try {
					datastoreResults = datastoreService.get(transaction, datastoreKeys);
					break;
				} catch (DatastoreTimeoutException datastoreTimeoutException) {
					// retry logic
					if (attempts == 5) throwIOWrappedException(datastoreTimeoutException, "read entities");;
					try {
						Thread.sleep(timeout);
					} catch (InterruptedException interruptedException) {}
					timeout *= 2;
					attempts ++;
				}
				
				// assuming the datastore results did come back, write them where they need to go
				if (datastoreResults != null) {
					// put the results into the return map
					entities.putAll(datastoreResults);
					
					// put the entities into memcache
					if (useMemcache) {
						// deal with the silly casting issue again
						Map<Object, Object> memcacheMap = new HashMap<Object, Object>();
						for (Object key : datastoreResults.keySet())
							memcacheMap.put(key, datastoreResults.get(key));
						try {
							// try to write all the entities to the memcache service
							memcacheService.putAll(memcacheMap);
						} catch (MemcacheServiceException memcacheServiceException) {
							// we couldn't store them at once, so lets try to write each one individually
							for (Object key : memcacheMap.keySet())
								try {
									// write the individual entity to the service
									memcacheService.put(key, memcacheMap.get(key));
								} catch (MemcacheServiceException memcacheServiceException2) {}
						} catch (OverQuotaException overQuotaException) {}
					}
				}
			}
		} catch (RuntimeException runtimeException) {
			throwIOWrappedException(runtimeException, "read entities");
		}
		
		return entities;
	}
	
	
	public static final Key writeEntityToPersistentStore(Entity entity) throws IOException {
		// default here is that we do write to the memcache
		return writeEntityToPersistentStore(entity, true);
	}
	
	public static final Key writeEntityToPersistentStore(Entity entity, boolean useMemcache) throws IOException {
		// default here is that oneTransaction is false, as most entity writes are in different entity groups; although not as much with the Lucene library
		// not that it matters to those calls as they will use the four parameter version
		return writeEntityToPersistentStore(entity, useMemcache, null);
	}
	
	public static final Key writeEntityToPersistentStore(Entity entity, boolean useMemcache, boolean oneTransaction) throws IOException {
		ArrayList<Entity> entities = new ArrayList<Entity>(1);
		entities.add(entity);
		return writeEntitiesToPersistentStore(entities, useMemcache, oneTransaction).get(0);
	}
	
	public static final Key writeEntityToPersistentStore(Entity entity, boolean useMemcache, Transaction transaction) throws IOException {
		// default here is the trySingleWrite is based on oneTransaction
		ArrayList<Entity> entities = new ArrayList<Entity>(1);
		entities.add(entity);
		return writeEntitiesToPersistentStore(entities, useMemcache, transaction, true).get(0);
	}
	
	public static final List<Key> writeEntitiesToPersistentStore(List<Entity> entities) throws IOException {
		// default here is that we do write to the memcache
		return writeEntitiesToPersistentStore(entities, true);
	}
	
	public static final List<Key> writeEntitiesToPersistentStore(List<Entity> entities, boolean useMemcache) throws IOException {
		// default here is that oneTransaction is false, as most entity writes are in different entity groups; although not as much with the Lucene library
		// not that it matters to those calls as they will use the four parameter version
		return writeEntitiesToPersistentStore(entities, useMemcache, null);
	}
	
	public static final List<Key> writeEntitiesToPersistentStore(List<Entity> entities, boolean useMemcache, boolean oneTransaction) throws IOException {
		return writeEntitiesToPersistentStore(entities, useMemcache, oneTransaction, true);
	}
	
	public static final List<Key> writeEntitiesToPersistentStore(List<Entity> entities, boolean useMemcache, Transaction transaction) throws IOException {
		// default here is the trySingleWrite is based on oneTransaction
		return writeEntitiesToPersistentStore(entities, useMemcache, transaction, transaction != null);
	}
	
	
	public static final List<Key> writeEntitiesToPersistentStore(List<Entity> entities, boolean useMemcache, boolean oneTransaction, boolean trySinglePut) throws IOException {
		Transaction trx = null;
		try {
			trx = oneTransaction ? DatastoreServiceFactory.getDatastoreService().beginTransaction() : null;
			List<Key> keys = writeEntitiesToPersistentStore(entities, useMemcache, trx, trySinglePut);
			attemptCommit(trx);
			return keys;
		} finally {
			checkTransaction(trx);
		}
	}
	
	/**
	 * Stores the entities using the key already assigned to the entities.
	 * 
	 * If the datastore issues a timeout error, the method will retry up to five times. With sleep times of 100ms, 200ms, 400ms, 800ms, 1600ms
	 * 
	 * @param entities - List of entities to be stored, having keys already is not required
	 * @param useMemcache - store the entities in memcache as well as the datastore, uses the entity key for a memcache key
	 * @Param transaction - transaction that will be used in calls datastore put calls
	 * @param trySinglePut - ask the method to attempt the store the entities using a single call to the datastore and/or memcache; failure will cause it to try again putting each individually
	 * @return Entity
	 * @throws IOException
	 */
	public static final List<Key> writeEntitiesToPersistentStore(List<Entity> entities, boolean useMemcache, Transaction transaction, boolean trySinglePut) throws IOException {
		DatastoreService datastoreService = DatastoreServiceFactory.getDatastoreService();
		MemcacheService memcacheService = useMemcache ? MemcacheServiceFactory.getMemcacheService() : null;
		if (memcacheService != null)
			memcacheService.setErrorHandler(memcacheErrorHandler);
		List<Key> returnKeys = null;
		
		// setup to write the entities in a repeat attempt loop 
		try {
			int timeout = 100;
			int attempts = 1;
			while (true) {
				// make sure that all crazy try/catch statements
				// don't create a infinite loop
				if (attempts == 6) break;
				
				try {
					try {
						if (trySinglePut || entities.size() == 1) {
							// store the entities in datastore
							returnKeys = datastoreService.put(transaction, entities);
						
							// store the entities in memcache
							if (useMemcache) {
								try {
									// use the putAll method if we have more than one entity
									if (entities.size() > 1) {
										// put the keys and entities into a HashMap
										HashMap<Object,Object> entitiesMap = new HashMap<Object,Object>(returnKeys.size());
										for (int i = 0; i < returnKeys.size(); i++)
											entitiesMap.put(returnKeys.get(i), entities.get(i));
										// store the HashMap
										memcacheService.putAll(entitiesMap);
									} else {
										// use the individual put if there is just one, save creating the HashMap
										memcacheService.put(returnKeys.get(0), entities.get(0));
									}
								} catch (MemcacheServiceException memcacheServiceException) {}
							}
						} else {
							// create the return key list
							returnKeys = new ArrayList<Key>(entities.size());
							// loop over the entities inserting them
							for (Entity entity : entities)
								returnKeys.add(datastoreService.put(transaction, entity));
							// store the entities in memcache
							if (useMemcache) {
								try {
									// loop over the keys and entities inserting them into the memcache
									for (int i = 0; i < returnKeys.size(); i++)
										memcacheService.put(returnKeys.get(i), entities.get(i));
								} catch (MemcacheServiceException memcacheServiceException) {}
							}
						}
						
						// exit the loop
						break;
					} catch (RequestTooLargeException requestTooLargeException) 
					{
						if (trySinglePut)
							// cache the state so we don't try again
							trySinglePut = false;
						else
							// pass the requestTooLargeException back to the caller
							throwIOWrappedException(requestTooLargeException, "write entities");					
					}

					// deal with the time exception by trying again
				} catch (DatastoreTimeoutException datastoreTimeoutException) {
					if (attempts == 5) throwIOWrappedException(datastoreTimeoutException, "write entities");
					try {
						Thread.sleep(timeout);
					} catch (InterruptedException interruptedException) {}
					timeout *= 2;
					attempts ++;
				}
			}

		} catch (RuntimeException runtimeException) {
			throwIOWrappedException(runtimeException, "write entities");
		}
		
		return returnKeys;
	}
	
	/**
	 * Attempts to commit the passed in transaction. Handles exceptions using the
	 * throwIOWrappedException method. 
	 * 
	 * @param transaction
	 * @throws IOException
	 */
	public static void attemptCommit(Transaction transaction) throws IOException {
		try {
			if (transaction != null && transaction.isActive()) transaction.commit();
		} catch (RuntimeException runtimeException) {
			throwIOWrappedException(runtimeException, "commit transaction");
		}
	}
	
	/**
	 * This method returns true when the versions do match, otherwise it returns false.
	 * Will throw IllegalArgumentException if version info cannot be found in the passed entity.
	 * 
	 * @param entity
	 * @return
	 * @throws IllegalArgumentException
	 */
	public static boolean isVersionEqual(Entity entity, long version) {
		Long passedVersion = getDirectoryVersion(entity);
		if (passedVersion == null) throw new IllegalArgumentException("No version information found in entity");
		return (passedVersion.equals(version));
	}
	
	/**
	 * It calls the wrapInIOException method, then throws the return value from that call. 
	 * If the wrapInIOException returns null, then the passed in exception is re-thrown.
	 * 
	 * IOException to the caller.
	 */
	public static void throwIOWrappedException(RuntimeException innerException, String operationText) throws IOException {
		IOException ioException = wrapInIOException(innerException, operationText);
		if (ioException != null)
			throw ioException;
		else throw innerException;
	}
	
	/**
	 * This method will wrap a variety of common GAE specific exceptions in an IOException
	 * and include a description to go with the wrapped exception. The exception is returned then.
	 * 
	 * If the innerException is not an instanceof one of the exceptions we are looking for, then
	 * null will be returned by this method. Call code should probably handle other exceptions
	 * as they see fit, but should not assume this will handle all exception types.
	 * 
	 * @param innerException
	 * @param operationText
	 * @return
	 */
	public static IOException wrapInIOException(RuntimeException innerException, String operationText) {
		if (innerException == null)
			return null;
		if (operationText == null)
			operationText = "";
			
		if (innerException instanceof DatastoreFailureException)
			return new IOException("Datastore failure while trying to " + operationText, innerException);
		else if (innerException instanceof DatastoreTimeoutException)
			return new IOException("Datastore timeout while trying to " + operationText, innerException);
		else if (innerException instanceof RequestTooLargeException)
			return new IOException("API request generated by operation [" + operationText + "] was too large", innerException);
		else if (innerException instanceof CapabilityDisabledException)
			return new IOException("Datastore is in read-only maintenance mode, can't " + operationText, innerException);
		else if (innerException instanceof OverQuotaException)
			return new IOException("The service has used up it's quota while trying to " + operationText, innerException);
		else if (innerException instanceof ApiProxyException)
			return new IOException("ApiProxyException while trying to " + operationText, innerException);
		else return null;
	}
	
	/**
	 * This method will check if the passed transaction is active. If it is,
	 * the method will call a rollback and throw a TransactionRolledbackException.
	 * 
	 * If null is passed in, the method will call the datastoreService.getCurrentTransaction(null) method,
	 * to see if there is an active transaction on this thread.
	 * 
	 * @param trx
	 */
	public static void checkTransaction(Transaction trx) throws TransactionRolledbackException {
		try {
		if (trx == null ) trx = DatastoreServiceFactory.getDatastoreService().getCurrentTransaction(null);
		if (trx != null && trx.isActive()) {
			trx.rollback();
			throw new TransactionRolledbackException(trx.toString());	
		}
		// we are going to ignore several possible exceptions that we can't recover from, that
		// probably happened before this was called in a finally clause.
		} catch (RuntimeException runtimeException) {}
	}
	
	/**
	 * This method is designed so that if two semi-independent operations fail, we the information from
	 * both exceptions can be combined into a single message, then thrown in a new IOException.
	 * 
	 * If all passed exceptions are null, then nothing happens; if any are not null then they are put 
	 * into the new IOException and thrown.
	 * 
	 * If the logger is not null, the message is written to the logger using the severe() method.
	 * 
	 * @link {DatastoreLockFactory} for an example of this, where the hookDirectory is used.
	 * 
	 * @param logger
	 * @param ioExceptions
	 * @throws IOException
	 */
	public static void combineMultipleIOExceptions(Logger logger, IOException... ioExceptions) throws IOException {
		StringBuilder builder = new StringBuilder();
		boolean foundOne = false;
		
		for (IOException ioException : ioExceptions)
			if (ioException != null) {
				foundOne = true;
				
				builder.append(ioException.getClass());
				builder.append(": ");
				builder.append(ioException.getMessage());
				builder.append('\n');
				Throwable cause = ioException.getCause(); 
				if (cause != null) {
					builder.append(cause.getClass());
					builder.append(": ");
					builder.append(cause.getMessage());
					builder.append('\n');
				}
			}
			
		if (foundOne) {
			String message = builder.toString();
			if (logger != null) logger.severe(message);
			throw new IOException(message);
		}
	}
}
